/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.nutch.indexwriter.song;

import org.apache.hadoop.conf.Configuration;
import org.apache.lucene.document.Document;
import org.apache.lucene.document.Field;
import org.apache.lucene.index.Term;
import org.apache.nutch.indexer.CleaningJob;
import org.apache.nutch.indexer.IndexWriter;
import org.apache.nutch.indexer.NutchDocument;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.ArrayList;
import java.util.List;
import java.util.Map.Entry;

public class SongIndexWriter implements IndexWriter {

    public static final Logger LOG = LoggerFactory
            .getLogger(SongIndexWriter.class);

    private Configuration config;

    private final List<Document> inputDocs = new ArrayList<Document>();

    private int batchSize = 200;
    private int numDeletes = 0;
    private boolean delete = false;

    private String directoryPath = null;

    protected static long documentCount = 0;

    @Override
    public void open(Configuration conf) throws IOException {
        batchSize = conf.getInt(SongConstants.COMMIT_SIZE, 200);
    }

    /**
     * @author Joycessh
     * @param doc
     * @return void
     * @date 5/1/19 3:27 PM
     */
    @Override
    public void write(NutchDocument doc) throws IOException {
        final Document inputDoc = new Document();

//        LOG.info(printDocumentInfo(doc));
        /**
         * NuthDocument:
         * tstamp : 1970-01-01T00:00:00.000Z
         * anchor : Aller au contenu
         * 	digest : 4fad4a9379c50019afa1fc492fa1c7a2
         * 	host : joycessh.xyz
         * 	boost : 2.828427
         * 	id : xyz.joycessh:http/
         * 	title : Joycessh
         * 	url : http://joycessh.xyz/
         * 	content : Joycessh 天若 有情 天亦老 我为 长者 续一秒
         */
        for (final Entry<String, List<String>> e : doc) {
            for (final String val : e.getValue()) {
                if (e.getKey().equals("content") || e.getKey().equals("title")) {
                    inputDoc.add(new Field(e.getKey(), val, FieldTypeUtils.getIndexedTermVectorsStoredType()));
                } else {
                    inputDoc.add(new Field(e.getKey(), val, FieldTypeUtils.getIndexedStoredType()));
                }
            }
        }
        inputDocs.add(inputDoc);
        documentCount++;
        if (inputDocs.size() >= batchSize) {
            try {
                LOG.info("Adding " + Integer.toString(inputDocs.size()) + " documents");
                IndexWriterUtil.getInstance().addDocuments(inputDocs);
            } catch (final IOException e) {
                throw new IOException(e);
            }
            inputDocs.clear();
        }
    }

    @Override
    public void close() throws IOException {
        try {
            if (!inputDocs.isEmpty()) {
                LOG.info("Adding " + Integer.toString(inputDocs.size()) + " documents");
                IndexWriterUtil.getInstance().addDocuments(inputDocs);
                inputDocs.clear();
            } else if (numDeletes > 0) {
                LOG.info("Deleted " + Integer.toString(numDeletes) + " documents");
            }
        } catch (final IOException e) {
            throw new IOException(e);
        }
    }

    @Override
    public Configuration getConf() {
        return config;
    }

    @Override
    public void setConf(Configuration conf) {
        config = conf;
        String directoryPath = SongConstants.DIRECTORY_PATH;
        delete = config.getBoolean(CleaningJob.ARG_COMMIT, false);
        if (directoryPath == null) {
            String message = "Missing DIRECTORY PATH. Should be set in song.conf "
                    + SongConstants.DIRECTORY_PATH;
            message += "\n" + describe();
            LOG.error(message);
            throw new RuntimeException(message);
        }
    }

    @Override
    public void delete(String key) throws IOException {
        if (delete) {
            try {
                Term term = new Term("id", key);
                IndexWriterUtil.getInstance().deleteDocuments(term);
                numDeletes++;
            } catch (final IOException e) {
                throw e;
            }
        }
    }

    @Override
    public void update(NutchDocument doc) throws IOException {
        write(doc);
    }

    @Override
    public void commit() throws IOException {
        try {
            IndexWriterUtil.getInstance().commit();
            LOG.info("Total " + documentCount
                    + (documentCount > 1 ? " documents are " : " document is ")
                    + "added.");
        } catch (IOException e) {
            throw e;
        }
    }

    @Override
    public String describe() {
        StringBuffer sb = new StringBuffer("SONGIndexWriter\n");
        sb.append("\t").append(SongConstants.DIRECTORY_PATH)
                .append(" : PATH of the Lucene Directory File\n");
        sb.append("\t").append(SongConstants.COMMIT_SIZE)
                .append(" : buffer size when sending to SONG (default 1000)\n");
        return sb.toString();
    }

    /**
     * 输出NuthDocument的信息
     * @author Joycessh
     * @param doc
     * @return java.lang.String
     * @date 5/4/19 1:40 PM
     */
    public String printDocumentInfo(NutchDocument doc) {
        StringBuffer sb = new StringBuffer("NuthDocument\n");
        for (final Entry<String, List<String>> e : doc) {
            for (final String val : e.getValue()) {

                sb.append("\t").append(e.getKey())
                        .append(" : " + val + "\n");
            }
        }
        return sb.toString();
    }

}
